/** 
 * @Title:扩展quartz.MethodInvokingJobDetailFactoryBean，加入对zk监控判断当前节点为master时，才确实执行job  
 * @Desription:扩展quartz.MethodInvokingJobDetailFactoryBean，加入对zk监控判断当前节点为master时，才确实执行job
 * @Company:CSN
 * @ClassName:MethodInvokingJobDetailFactoryBean.java
 * @Author:Justic
 * @CreateDate:2015-7-1   
 * @UpdateUser:Justic  
 * @Version:0.1 
 *    
 */

package com.justic.zk.quartz.quartz;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.Scheduler;
import org.quartz.StatefulJob;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeanWrapper;
import org.springframework.beans.PropertyAccessorFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.support.ArgumentConvertingMethodInvoker;
import org.springframework.scheduling.quartz.JobMethodInvocationFailedException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.MethodInvoker;
import org.springframework.util.ReflectionUtils;

import com.justic.zk.quartz.ScheduleManager;

/**
 * @ClassName: MethodInvokingJobDetailFactoryBean
 * @Description:扩展quartz.MethodInvokingJobDetailFactoryBean，加入对zk监控判断当前节点为master时，才确实执行job
 * @author: Justic
 * @date: 2015-7-1
 * 
 */
@SuppressWarnings({ "deprecation", "rawtypes", "unchecked" })
public class MethodInvokingJobDetailFactoryBean extends ArgumentConvertingMethodInvoker implements
		FactoryBean<JobDetail>, BeanNameAware, BeanClassLoaderAware, BeanFactoryAware, InitializingBean {
	private static Class<?> jobDetailImplClass;

	private static Method setResultMethod;

	// zk
	private static ScheduleManager zkSchedule;

	static {
		try {
			jobDetailImplClass = Class.forName("org.quartz.impl.JobDetailImpl");
		} catch (ClassNotFoundException ex) {
			jobDetailImplClass = null;
		}
		try {
			Class jobExecutionContextClass = QuartzJobBean.class.getClassLoader().loadClass(
					"org.quartz.JobExecutionContext");
			setResultMethod = jobExecutionContextClass.getMethod("setResult", Object.class);
		} catch (Exception ex) {
			throw new IllegalStateException("Incompatible Quartz API: " + ex);
		}
	}

	private String name;

	private String group = Scheduler.DEFAULT_GROUP;

	private boolean concurrent = true;

	private String targetBeanName;

	private String[] jobListenerNames;

	private String beanName;

	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();

	private BeanFactory beanFactory;

	private JobDetail jobDetail;

	/**
	 * Set the name of the job.
	 * <p>
	 * Default is the bean name of this FactoryBean.
	 * 
	 * @see org.quartz.JobDetail#setName
	 */
	public void setName(String name) {
		this.name = name;
	}

	/**
	 * Set the group of the job.
	 * <p>
	 * Default is the default group of the Scheduler.
	 * 
	 * @see org.quartz.JobDetail#setGroup
	 * @see org.quartz.Scheduler#DEFAULT_GROUP
	 */
	public void setGroup(String group) {
		this.group = group;
	}

	/**
	 * Specify whether or not multiple jobs should be run in a concurrent
	 * fashion. The behavior when one does not want concurrent jobs to be
	 * executed is realized through adding the {@link StatefulJob} interface.
	 * More information on stateful versus stateless jobs can be found <a href=
	 * "http://www.quartz-scheduler.org/documentation/quartz-2.1.x/tutorials/tutorial-lesson-03"
	 * >here</a>.
	 * <p>
	 * The default setting is to run jobs concurrently.
	 */
	public void setConcurrent(boolean concurrent) {
		this.concurrent = concurrent;
	}

	/**
	 * Set the name of the target bean in the Spring BeanFactory.
	 * <p>
	 * This is an alternative to specifying {@link #setTargetObject
	 * "targetObject"}, allowing for non-singleton beans to be invoked. Note
	 * that specified "targetObject" and {@link #setTargetClass "targetClass"}
	 * values will override the corresponding effect of this "targetBeanName"
	 * setting (i.e. statically pre-define the bean type or even the bean
	 * object).
	 */
	public void setTargetBeanName(String targetBeanName) {
		this.targetBeanName = targetBeanName;
	}

	/**
	 * Set a list of JobListener names for this job, referring to non-global
	 * JobListeners registered with the Scheduler.
	 * <p>
	 * A JobListener name always refers to the name returned by the JobListener
	 * implementation.
	 * 
	 * @see SchedulerFactoryBean#setJobListeners
	 * @see org.quartz.JobListener#getName
	 */
	public void setJobListenerNames(String[] names) {
		this.jobListenerNames = names;
	}

	public void setBeanName(String beanName) {
		this.beanName = beanName;
	}

	public void setBeanClassLoader(ClassLoader classLoader) {
		this.beanClassLoader = classLoader;
	}

	public void setBeanFactory(BeanFactory beanFactory) {
		this.beanFactory = beanFactory;
	}

	@Override
	protected Class resolveClassName(String className) throws ClassNotFoundException {
		return ClassUtils.forName(className, this.beanClassLoader);
	}

	@SuppressWarnings("unused")
	public void afterPropertiesSet() throws ClassNotFoundException, NoSuchMethodException {
		prepare();

		// Use specific name if given, else fall back to bean name.
		String name = (this.name != null ? this.name : this.beanName);

		// Consider the concurrent flag to choose between stateful and stateless
		// job.
		Class jobClass = (this.concurrent ? MethodInvokingJob.class : StatefulMethodInvokingJob.class);

		// Build JobDetail instance.
		if (jobDetailImplClass != null) {
			// Using Quartz 2.0 JobDetailImpl class...
			this.jobDetail = (JobDetail) BeanUtils.instantiate(jobDetailImplClass);
			BeanWrapper bw = PropertyAccessorFactory.forBeanPropertyAccess(this.jobDetail);
			bw.setPropertyValue("name", name);
			bw.setPropertyValue("group", this.group);
			bw.setPropertyValue("jobClass", jobClass);
			bw.setPropertyValue("durability", true);
			((JobDataMap) bw.getPropertyValue("jobDataMap")).put("methodInvoker", this);
		}
		/*
		 * 依赖quartz 2.x 所以mark掉
		 * else { // Using Quartz 1.x JobDetail class... this.jobDetail = new
		 * JobDetail(name, this.group, jobClass);
		 * this.jobDetail.setVolatility(true);
		 * this.jobDetail.setDurability(true);
		 * this.jobDetail.getJobDataMap().put("methodInvoker", this); }
		 */

		// Register job listener names.
		if (this.jobListenerNames != null) {
			for (String jobListenerName : this.jobListenerNames) {
				if (jobDetailImplClass != null) {
					throw new IllegalStateException("Non-global JobListeners not supported on Quartz 2 - "
							+ "manually register a Matcher against the Quartz ListenerManager instead");
				}
				// this.jobDetail.addJobListener(jobListenerName);
			}
		}

		postProcessJobDetail(this.jobDetail);
	}

	/**
	 * Callback for post-processing the JobDetail to be exposed by this
	 * FactoryBean.
	 * <p>
	 * The default implementation is empty. Can be overridden in subclasses.
	 * 
	 * @param jobDetail
	 *            the JobDetail prepared by this FactoryBean
	 */
	protected void postProcessJobDetail(JobDetail jobDetail) {
	}

	/**
	 * Overridden to support the {@link #setTargetBeanName "targetBeanName"}
	 * feature.
	 */
	@Override
	public Class getTargetClass() {
		Class targetClass = super.getTargetClass();
		if (targetClass == null && this.targetBeanName != null) {
			Assert.state(this.beanFactory != null, "BeanFactory must be set when using 'targetBeanName'");
			targetClass = this.beanFactory.getType(this.targetBeanName);
		}
		return targetClass;
	}

	/**
	 * Overridden to support the {@link #setTargetBeanName "targetBeanName"}
	 * feature.
	 */
	@Override
	public Object getTargetObject() {
		Object targetObject = super.getTargetObject();
		if (targetObject == null && this.targetBeanName != null) {
			Assert.state(this.beanFactory != null, "BeanFactory must be set when using 'targetBeanName'");
			targetObject = this.beanFactory.getBean(this.targetBeanName);
		}
		return targetObject;
	}

	public JobDetail getObject() {
		return this.jobDetail;
	}

	public Class<? extends JobDetail> getObjectType() {
		return (this.jobDetail != null ? this.jobDetail.getClass() : JobDetail.class);
	}

	public boolean isSingleton() {
		return true;
	}

	/**
	 * Quartz Job implementation that invokes a specified method. Automatically
	 * applied by MethodInvokingJobDetailFactoryBean.
	 */
	public static class MethodInvokingJob extends QuartzJobBean {

		protected static final Log logger = LogFactory.getLog(MethodInvokingJob.class);

		private MethodInvoker methodInvoker;

		/**
		 * Set the MethodInvoker to use.
		 */
		public void setMethodInvoker(MethodInvoker methodInvoker) {
			this.methodInvoker = methodInvoker;
		}

		/**
		 * Invoke the method via the MethodInvoker.
		 */
		@Override
		protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
			try {
				//新增zookeeper-leader选举，实现多节点的高可用
				if (zkSchedule.start()) {
					// 加入zk判断当前是否为leader，才可执行
					if (zkSchedule.getCurator().isLeader()) {

						ReflectionUtils.invokeMethod(setResultMethod, context, this.methodInvoker.invoke());
					}
				}else{
					logger.error("connect zk-server failure!!!!");
				}

			} catch (InvocationTargetException ex) {
				if (ex.getTargetException() instanceof JobExecutionException) {
					// -> JobExecutionException, to be logged at info level by
					// Quartz
					throw (JobExecutionException) ex.getTargetException();
				} else {
					// -> "unhandled exception", to be logged at error level by
					// Quartz
					throw new JobMethodInvocationFailedException(this.methodInvoker, ex.getTargetException());
				}
			} catch (Exception ex) {
				// -> "unhandled exception", to be logged at error level by
				// Quartz
				throw new JobMethodInvocationFailedException(this.methodInvoker, ex);
			}
		}
	}

	/**
	 * Extension of the MethodInvokingJob, implementing the StatefulJob
	 * interface. Quartz checks whether or not jobs are stateful and if so,
	 * won't let jobs interfere with each other.
	 */
	public static class StatefulMethodInvokingJob extends MethodInvokingJob implements StatefulJob {

		// No implementation, just an addition of the tag interface StatefulJob
		// in order to allow stateful method invoking jobs.
	}

	public ScheduleManager getZkSchedule() {
		return zkSchedule;
	}

	@SuppressWarnings("static-access")
	public void setZkSchedule(ScheduleManager zkSchedule) {
		this.zkSchedule = zkSchedule;
	}
}
